package com.example

import java.sql.PreparedStatement
import com.alibaba.druid.pool.{DruidDataSource, DruidPooledConnection}
import org.apache.spark.sql.ForeachWriter

class MysqlWriter extends ForeachWriter[BusInfo] {
  var connection:DruidPooledConnection = _
  var preparedStatement: PreparedStatement = _

  override def open(partitionId: Long, epochId: Long): Boolean = {
    connection = MysqlWriter.getConnection()
    if (connection != null) true
    else false
  }

  override def process(value: BusInfo): Unit = {
    val lglat: String = value.lglat
    val deployNum: String = value.deployNum
    val sql = "update businfo set lglat=? where deployNum=?"
    preparedStatement = connection.prepareStatement(sql)
    preparedStatement.setString(1,lglat)
    preparedStatement.setString(2,deployNum)
    val result: Int = preparedStatement.executeUpdate()
    if (result > 0) println("修改成功")
    else println("修改失败")

  }

  override def close(errorOrNull: Throwable): Unit = {
    preparedStatement.close()
    connection.close()
  }
}

object MysqlWriter {
  //数据源配置
  private val dataSource:DruidDataSource = new DruidDataSource()


  def getConnection():DruidPooledConnection = {
    dataSource.setUrl("jdbc:mysql://linux123:3306/test?useSSL=false");
    dataSource.setUsername("root");
    dataSource.setPassword("12345678");

    //下面都是可选的配置
    dataSource.setInitialSize(10);  //初始连接数，默认0
    dataSource.setMaxActive(30);  //最大连接数，默认8
    dataSource.setMinIdle(10);  //最小闲置数
    dataSource.setMaxWait(2000);  //获取连接的最大等待时间，单位毫秒
    dataSource.setPoolPreparedStatements(true); //缓存PreparedStatement，默认false
    //缓存PreparedStatement的最大数量，默认-1（不缓存）。
    // 大于0时会自动开启缓存PreparedStatement，所以可以省略上一句代码
    dataSource.setMaxOpenPreparedStatements(20);
    //获取连接
    dataSource.getConnection();
  }
}